package com.athui.service;

import com.athui.bean.metadata.ColumnMetaData;
import com.athui.bean.task.FileTask;
import com.athui.bean.task.SameTask;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.io.FileReader;
import java.io.IOException;
import java.io.LineNumberReader;

/**
 * @description: TODO
 *  文件最后一行不能有空行
 * @autor: zhangzhonghui
 * @create: 2024-08-21 17:36
 * @Version: 1.0
 */
public abstract class FileExplorerService<OUT> extends ExplorerServer<String,OUT> {

    private FileTask fileTask;

    public FileExplorerService(FileTask fileTask) {
        this.fileTask = fileTask;
    }

    @Override
    public void open(StreamExecutionEnvironment env) {}

    @Override
    public Integer getRowSizeLimit(){
        try (LineNumberReader lineNumberReader = new LineNumberReader(new FileReader(fileTask.getFilePath()))){
            lineNumberReader.skip(Long.MAX_VALUE);
            int lineNumber = lineNumberReader.getLineNumber();
            //实际上是读取换行符数量 , 所以需要+1
            return lineNumber + 1;
        } catch (IOException e) {
           e.printStackTrace();
        }
        return 0;
    }

    @Override
    public DataStream<String> source(StreamExecutionEnvironment env) {
        FileSource<String> fileSource = FileSource.forRecordStreamFormat(new TextLineInputFormat(fileTask.getCharsetName()), new Path(fileTask.getFilePath())).build();
        return env.fromSource(fileSource, WatermarkStrategy.noWatermarks(), "file-source");
    }

    @Override
    public void sink(DataStream<Tuple2<ColumnMetaData, SameTask>> stream) {
        stream.print();
    }

    @Override
    public String getJobName() {
        return "file-job-name";
    }

    @Override
    public void run() throws Exception {
        super.run();
    }



}
